/*
 * Package il.ac.biu.cs.grossmm.impl.cpp
 * File InmemoryProvider.java
 * 
 * This is a part of presence service framework prototype. 
 * See javadoc for more information.
 *  
 * Copyright (C) 2006 Grossmann Mark
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public License
 * as published by the Free Software Foundation; either version 2
 * of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.

 * You should have received a copy of the GNU Lesser General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 */
package il.ac.biu.cs.grossmm.impl.cpp;

import static il.ac.biu.cs.grossmm.api.data.NodeOf.*;
import static il.ac.biu.cs.grossmm.api.keys.KeyFactory.*;
import static il.ac.biu.cs.grossmm.api.keys.PatternFactory.*;
import static il.ac.biu.cs.grossmm.api.presence.BaseVocabulary.*;
import il.ac.biu.cs.grossmm.api.OperationFailedException;
import il.ac.biu.cs.grossmm.api.cpp.CppServiceConsumer;
import il.ac.biu.cs.grossmm.api.cpp.CppServiceProvider;
import il.ac.biu.cs.grossmm.api.cpp.State;
import il.ac.biu.cs.grossmm.api.data.Root;
import il.ac.biu.cs.grossmm.api.data.NodeOf;
import il.ac.biu.cs.grossmm.api.flow.ActiveDataManager;
import il.ac.biu.cs.grossmm.api.flow.NodeEventInspector;
import il.ac.biu.cs.grossmm.api.flow.NodeInspector;
import il.ac.biu.cs.grossmm.api.flow.Subscriber;
import il.ac.biu.cs.grossmm.api.flow.SubscriptionPoint;
import il.ac.biu.cs.grossmm.api.flow.Status;
import il.ac.biu.cs.grossmm.api.keys.Key;
import il.ac.biu.cs.grossmm.api.keys.KeyPattern;
import il.ac.biu.cs.grossmm.api.presence.Entity;

import java.util.HashMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

public class InmemoryProvider<T> implements CppServiceProvider<T>,
	Subscriber<NodeOf<T>> {
	protected final ActiveDataManager adm;

	protected final CppServiceConsumer<T> consumer;

	protected final Class<T> tClass;

	protected SubscriptionPoint<NodeOf<T>> sp;

	HashMap<Key, Copies<T>> byKey = new HashMap<Key, Copies<T>>();

	HashMap<String, Subscription> byId = new HashMap<String, Subscription>();

	DelayQueue<Subscription> delayQueue = new DelayQueue<Subscription>();

	ReentrantLock opLock = new ReentrantLock();

	static class Copies<T> extends HashMap<String, Subscription<T>> {
		/**
		 * 
		 */
		private static final long serialVersionUID = 3004034925342030167L;

		boolean inProcess = true;

		boolean isSuccess;

		final Entity watcher;

		final Key key;

		final Key target;

		T value;

		/**
		 * @param value
		 * @param status
		 */
		public Copies(Entity watcher, Key key, Key target)
		{
			this.watcher = watcher;
			this.key = key;
			this.target = target;
		}
	}

	static class Subscription<T> implements Delayed {
		final String subscriptID;

		final Copies<T> copies;

		long validUntil;

		int duration;

		/**
		 * @param subscriptID
		 * @param watcher
		 * @param target
		 * @param status
		 * @param duration
		 */
		public Subscription(Copies<T> copies, String subscriptID, int duration)
		{
			this.copies = copies;
			this.subscriptID = subscriptID;
			this.duration = duration;
			this.validUntil = System.currentTimeMillis() + duration * 1000;
		}

		int getDuration()
		{
			long left = validUntil - System.currentTimeMillis();

			if (left < 0)
				return 0;

			return (int) (left / 1000);
		}

		public long getDelay(TimeUnit unit)
		{
			return unit.convert(getDuration(), TimeUnit.SECONDS);
		}

		public int compareTo(Delayed o)
		{
			long l = getDuration() - o.getDelay(TimeUnit.MILLISECONDS);

			// to avoid long->int
			if (l > 0)
				return 1;
			if (l < 0)
				return -1;
			return 0;
		}
	}

	/**
	 * @param adm
	 */
	public InmemoryProvider(final ActiveDataManager adm, KeyPattern pattern,
		Class<T> tClass, CppServiceConsumer<T> consumer)
	{
		super();
		this.adm = adm;
		this.consumer = consumer;
		this.tClass = tClass;
		
		KeyPattern pattern2 = insert(pattern, mandatory(WATCHER, Entity.class));

		sp = adm.createSubscriptionPoint(this, pattern2, nodeOf(tClass));
	}

	public State subscribe(Entity watcher, Key target, String subscriptID,
		int duration)
	{
		Key key = insert(target, entry(WATCHER, watcher));

		boolean isCopy;
		boolean isRefresh;

		Subscription<T> subscription;
		Copies<T> copies;

		opLock.lock();
		try {
			copies = byKey.get(key);
			isCopy = (copies != null);
			if (!isCopy) {
				copies = new Copies<T>(watcher, key, target);
				byKey.put(key, copies);
			}
			
			synchronized(copies) {
				subscription = copies.get(subscriptID);
				
				if( subscription == null ) {
					isRefresh = false;
					subscription = new Subscription<T>(copies, subscriptID, duration);
					copies.put(subscriptID, subscription);
				}
					isRefresh = true;
			}

		} finally {
			opLock.unlock();
		}

		if (isCopy) {
			if( isRefresh ) {
				// refresh existing subscription (update duration)
				// TODO
			} else
				// backend subscription already active or in progress
				return waitBackendSubscription(subscription);
		}

		// backend subscription does not exist yet
		if ( sp.subscribe(key, null) == Status.UNMATCHED) {
			deactivateAll(key);
			return new State(false, 0);
		}

		return waitBackendSubscription(subscription);
	}

	private State waitBackendSubscription(Subscription<T> subscription)
	{
		State state;
		Copies<T> copies = subscription.copies;
		T value;

		synchronized (copies) {
			// if in progress - wait until ready
			while (copies.inProcess) {
				try {
					copies.wait();
				} catch (InterruptedException e) {
					return new State(false, 0);
				}
			}

			// backend subscription active
			state = new State(copies.isSuccess, subscription.getDuration());
			value = copies.value;

			if (!copies.isSuccess)
				copies.remove(subscription);
		}

		consumer.notify(copies.watcher, copies.target, state,
			subscription.subscriptID, value);

		return state;
	}

	public void activated(Key key, Root<NodeOf<T>> node)
	{
		opLock.lock();

		try {
			Copies<T> copies = byKey.get(key);
			synchronized (copies) {
				NodeInspector ins;
				try {
					ins = sp.readLock(node);
					copies.value =
						ins.getValue(node, nodeOf(tClass).PROPERTY);
					copies.isSuccess = true;
				} catch (OperationFailedException e) {
					System.out.println(e);
					sp.unsubscribe(key);
					copies.isSuccess = false;
					copies.value = null;

					removeCopy(copies);
				} finally {
					sp.readUnlock(node);
				}

				copies.inProcess = false;
				copies.notifyAll();
			}
		} finally {
			opLock.unlock();
		}
	}

	public void deactivated(Key key, Status status)
	{
		deactivateAll(key);
	}

	/**
	 * @param key
	 */
	private void deactivateAll(Key key)
	{
		Copies<T> copies;
		opLock.lock();
		 
		try {
			copies = byKey.get(key);

			if (copies == null) {
				assert false;
				return;
			}
			
			synchronized (copies) {
				copies.value = null;
				copies.inProcess = false;
				copies.isSuccess = false;
			}

			byKey.remove(key);
		} finally {
			opLock.unlock();
		}

		synchronized (copies) {
			copies.notifyAll();

			for (Subscription s : copies.values()) {
				State state = new State(false, 0);

				consumer.notify(copies.watcher, copies.target, state,
					s.subscriptID, null);
			}
		}
	}

	public void notify(Root<NodeOf<T>> node,
		NodeEventInspector inspector)
	{
		Key key = node.getKey();
		Copies<T> copies = byKey.get(key);

		T value;

		if (copies == null) {
			assert false;
			return;
		}

		try {
			value = inspector.getValue(node, property(tClass));

			synchronized (copies) {
				copies.value = value;
				copies.inProcess = false;
				copies.isSuccess = true;
			}

			for (Subscription s : copies.values() ) {
				State state = new State(true, s.getDuration());

				consumer.notify(copies.watcher, copies.target, state,
					s.subscriptID, value);
			}

		} catch (OperationFailedException e) {
			e.printStackTrace(); 
			deactivateAll(key);
		}
	}

	/**
	 * @param copies
	 */
	private void removeCopy(Copies copies)
	{
		opLock.lock();
		synchronized (copies) {
			copies.remove(copies);
			if (copies.isEmpty())
				byKey.remove(copies.key);
		}
		opLock.unlock();
	}
}
